package com.aotain.nyx.appstatd;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import com.aotain.nyx.statis.AbnStatisTuple;

public class AppAttStatisReduce implements ReduceFunction<Tuple2<String,AbnStatisTuple>> {

	/** 
	* @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	*/ 
	private static final long serialVersionUID = 8867731776636270995L;

	@Override
	public Tuple2<String, AbnStatisTuple> reduce(
			Tuple2<String, AbnStatisTuple> value1,
			Tuple2<String, AbnStatisTuple> value2) throws Exception {
		// TODO Auto-generated method stub
		AbnStatisTuple tuple = value1.f1;
		if(value1.f1 != null && value2.f1 != null ) {
			String time1 = value1.f1.getAttTime();
			String time2 = value2.f1.getAttTime();
			if(Long.parseLong(time2)>Long.parseLong(time1)) {
				tuple.setAttTime(time2);
			}else {
				tuple.setAttTime(time1);
			}
			tuple.setAttackNum(value1.f1.getAttackNum() + value2.f1.getAttackNum());
		}else {
			System.out.println("reduce arg is null");
		}
		return new Tuple2<String,AbnStatisTuple>(value1.f0,tuple);
	}

}
